package com.shujia.dws

import com.shujia.common.{DateUtil, SparkTool}
import com.shujia.util.Geography
import org.apache.spark.sql.expressions.{UserDefinedFunction, Window}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object DwsCityTouristMskDay extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 计算两个时间字符串的时间差（单位：s）
    val diff_date: UserDefinedFunction = udf((date_str1: String, date_str2: String) => {
      DateUtil.diff_date(date_str1, date_str2)
    })

    // 将计算两个网格之间的距离注册成UDF 才能够在SQL/DSL中使用
    val calculateLength: UserDefinedFunction = udf((grid_id1: String, grid_id2: String) => {
      Geography.calculateLength(grid_id1.toLong, grid_id2.toLong)
    })

    // 加载行政区域配置维度表
    val adminCodeDF: DataFrame = spark.table("dim.dim_admincode")

    // 加载位置数据融合表
    val mergeLocDF: DataFrame = spark.table("dwi.dwi_res_regn_mergelocation_msk_d").where($"day_id" === day_id)

    // 根据区县id获取城市id
    val mergeLocCityDF: DataFrame = mergeLocDF
      .join(adminCodeDF.hint("broadcast"), "county_id")
      .select($"mdn"
        , $"start_date"
        , $"end_date"
        , $"city_id"
        , $"longi"
        , $"lati"
        , $"bsid"
        , $"grid_id")

    // 加载用户画像维度表
    val userTagDF: DataFrame = spark.table("dim.dim_usertag_msk_d").where($"day_id" === day_id)


    /**
     * 市游客的定义：
     * 1、停留时间超过3小时
     * 2、出行距离大于10KM
     */
    val cityTouristDF: DataFrame = mergeLocCityDF
      // 计算每个用户在每个城市的停留时间
      .withColumn("min_start_date", min($"start_date") over Window.partitionBy($"mdn", $"city_id"))
      .withColumn("max_end_date", max($"end_date") over Window.partitionBy($"mdn", $"city_id"))
      .withColumn("d_stay_time", diff_date($"max_end_date", $"min_start_date"))
      .where($"d_stay_time" > 60 * 60 * 3)
      // 关联用户画像表
      .join(userTagDF, "mdn")
      // 通过居住地的网格id与目的地的网格id计算出行距离
      .withColumn("distance", calculateLength($"grid_id", $"resi_grid_id"))
      // 找到用户每一个居住地到每一个城市的最远距离 作为 出行距离
      .withColumn("d_max_distance", max($"distance") over Window.partitionBy($"mdn", $"resi_grid_id", $"city_id"))
      .where($"d_max_distance" > 10 * 1000)
      .select($"mdn", $"resi_county_id" as "source_county_id", $"city_id" as "d_city_id", $"d_stay_time", $"d_max_distance")
      .distinct()

    // 保存数据
    cityTouristDF
      .write
      .mode(SaveMode.Overwrite)
      .format("csv")
      .option("sep", ",")
      .save(s"/daas/motl/dws/dws_city_tourist_msk_d/day_id=$day_id")


    spark.sql(
      s"""
         |alter table dws.dws_city_tourist_msk_d add if not exists partition(day_id='$day_id')
         |""".stripMargin)

  }

}
